package com.afdor.rws.event.driven.channel.local;

import com.afdor.rws.annotation.extension.Extension;
import com.afdor.rws.core.extension.ExtensionLoader;
import com.afdor.rws.event.driven.channel.ChannelProvider;
import com.afdor.rws.event.driven.channel.DefaultChannelProvider;
import com.afdor.rws.event.driven.channel.consumer.EventConsumer;
import com.afdor.rws.event.driven.channel.consumer.handler.EventHandler;
import com.afdor.rws.event.driven.channel.consumer.handler.EventRetryStrategy;
import com.afdor.rws.event.driven.channel.local.observer.ObserverEventSource;
import com.afdor.rws.event.driven.channel.producer.EventProducer;
import com.afdor.rws.event.driven.config.EventConsumerConfig;
import com.afdor.rws.event.driven.config.EventProducerConfig;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.NonNull;

/**
 * 本地事件消息通道
 *
 * @author 悭梵
 * @date Created in 2018-08-08 10:38
 */
@NoArgsConstructor
@Extension(value = "local", order = Extension.NORMAL_ORDER - 100, scope = Extension.Scope.PROTOTYPE, interfaceClass = ChannelProvider.class)
public class LocalEventChannelProvider extends DefaultChannelProvider {

    @Getter
    private ObserverEventSource eventSource = new ObserverEventSource();

    @NonNull
    @Override
    protected EventConsumer createEventConsumer(@NonNull EventConsumerConfig eventConsumerConfig, @NonNull EventHandler eventHandler, EventRetryStrategy eventRetryStrategy) {
        /* 构建并设置消费者对象 */
        return ExtensionLoader
                .getInstance(EventConsumer.class)
                .newExtension(getConfig().getEventConsumerStrategyName(), getEventSource())
                .setEventHandler(eventHandler)
                .setEventRetryStrategy(eventRetryStrategy)
                .setSerialization(getSerialization())
                .setConfig(eventConsumerConfig)
                .setConsumerName(eventConsumerConfig.getConsumerName())
                .setEventSourceName(getEventSourceName());
    }

    @NonNull
    @Override
    protected EventProducer getEventProducer(EventProducerConfig eventProducerConfig) {
        if (!getEventProducerReference().isMarked()) {

            EventProducer eventProducer = ExtensionLoader
                    .getInstance(EventProducer.class)
                    .newExtension(getConfig().getEventProducerStrategyName(), getEventSource())
                    .setSerialization(getSerialization())
                    .setEventSourceName(getEventSourceName())
                    .setConfig(eventProducerConfig)
                    .init();

            getEventProducerReference().compareAndSet(null, eventProducer, false, true);
        }
        return getEventProducerReference().getReference();
    }


}
